package SparkStream

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.{SPARK_BRANCH, SparkConf}
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkKafka {

  def main(args: Array[String]): Unit = {
    /* spark streaming实现kafka 的 消费者
           1> 构建sparkconf 本地运行，运行应用程序名称
           2> 构建 sparkstreaming  ---> streamingContext ,加载配置
           3> kafka 配置 broker ， key value ,group id,消费模式
           4> spark 链接kafka 订阅，topic，streaming  contsxt
           5> 循环的方式 打印 / 处理
           6> 开启ssc ,监控kafka数据

     */
    //1> 构建sparkconf 本地运行，运行应用程序名称
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloSparkKafka")
      // StreamingContext 需导入依赖
    //spark streaming 可以进行流式处理
    val ssc = new StreamingContext(conf,Seconds(2))


    //spark 输出红色 info信息
    ssc.sparkContext.setLogLevel("error")


    //3> kafka 配置 broker ， key value ,group id,消费模式
    val kafkaParams = Map[String,Object](
      "bootstrap.servers" -> "192.168.134.128:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "niit",
      "enable.auto.commit" -> (false:java.lang.Boolean)
    )

   // 4> spark 链接kafka 订阅，topic，streaming  contsxt
   val topicName = Array("16test")
    val streamRdd = KafkaUtils.createDirectStream[String,String](
      ssc,
      PreferConsistent,
      Subscribe[String,String](topicName,kafkaParams)
    )

    //返回kafka 返回的streamRdd
    streamRdd.foreachRDD(
      x => {
        if(! x.isEmpty()){   //判断是否为空
          val line = x.map(_.value)

          line.foreach(println)
        }
      }
    )
    ssc.start()
    ssc.awaitTermination() //监控kafka




  }

}
